package local.blueheart.springcloud.rabbitmq.simple;

import com.rabbitmq.client.*;
import local.blueheart.springcloud.rabbitmq.common.BHCustomConsumer;
import local.blueheart.springcloud.rabbitmq.common.BHFailCustomConsumer;
import local.blueheart.springcloud.rabbitmq.util.MessageBrokerHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class SimpleQueueTest {


    private static final String QUEUE_NAME = "test_simple_queue";

    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleQueueTest.class);

    private MessageBrokerHelper messageBrokerHelper;

    public SimpleQueueTest() {
        String host = "127.0.0.1";
        int port = 5672;
        String vHost = "/bhmmr";
        String username = "bh_mmr";
        String password = "sunny6150";
        messageBrokerHelper = new MessageBrokerHelper(host,port,vHost,username,password);

    }

    public void sender(int senderCount,int delay){
        Connection connection = messageBrokerHelper.getConnection();
        Channel channel = messageBrokerHelper.getChannel(connection,QUEUE_NAME);
        try {
            int senderTime = senderCount < 1 ? 1 : senderCount;
            for (int i=0;i<senderTime;i++) {
                String body = "hello,rabbit mq" + i;
                LOGGER.debug("P_"+body);
                channel.basicPublish("", QUEUE_NAME, null, body.getBytes());
                Thread.sleep(delay);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            messageBrokerHelper.closeChannel(channel);
            messageBrokerHelper.closeConnection(connection);
        }
    }

    public void sender(int senderCount){
        Connection connection = messageBrokerHelper.getConnection();
        Channel channel = messageBrokerHelper.getChannel(connection,QUEUE_NAME);
        try {
            int senderTime = senderCount < 1 ? 1 : senderCount;
            for (int i=0;i<senderTime;i++) {
                String body = "hello,rabbit mq" + i;
                channel.basicPublish("", QUEUE_NAME, null, body.getBytes());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            messageBrokerHelper.closeChannel(channel);
            messageBrokerHelper.closeConnection(connection);
        }

    }

    public void failDisppathConsumer(String name,long delayTime){
        Connection connection = messageBrokerHelper.getConnection();
        Channel channel = messageBrokerHelper.getChannel(connection,QUEUE_NAME);
        Consumer consumer = new BHFailCustomConsumer(channel,delayTime,name);
        try {
            LOGGER.info("Consumer {} waiting consume,delay time {}",name,delayTime);
            channel.basicConsume(QUEUE_NAME,false,consumer);
        } catch (IOException e) {
            LOGGER.debug("consumer listener failure",e);
        }
    }

    public void consumer(String name)  {
        Connection connection = messageBrokerHelper.getConnection();
        Channel channel = messageBrokerHelper.getChannel(connection,QUEUE_NAME);
        Consumer consumer = new BHCustomConsumer(channel,name);
        try {
            channel.basicConsume(QUEUE_NAME,true,consumer);
        } catch (IOException e) {
            LOGGER.debug("consumer listener failure",e);
        }
    }

    public void consumer(){
        Connection connection = messageBrokerHelper.getConnection();
        Channel channel = messageBrokerHelper.getChannel(connection,QUEUE_NAME);
        Consumer basicConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                LOGGER.debug("consumer " + message);
            }
        };
        try {
            channel.basicConsume(QUEUE_NAME,true,basicConsumer);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args){
//        SimpleQueueTest simpleQueueTest = new SimpleQueueTest();
//        simpleQueueTest.sender(1000);
        SimpleQueueTest simpleQueueTest1 = new SimpleQueueTest();
        simpleQueueTest1.consumer();
    }



}
